// -*-c++-*-

#include <unistd.h>

#include <new>
#include <iostream>
#include <algorithm>
#include <limits>
#include <cassert>

#include <boost/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/foreach.hpp>

#include <xthread/guard.hpp>

#include <xbase_exception.hpp>
#include <xbase_utility.hpp>
#include <kv.hpp>

#include "global.hpp"
#include "table_schema.hpp"
#include "table_schema_manager.hpp"

#include "log_thread.hpp"

// 用于在子表映射列表中找出最老元素的小于比较器，只比较第二个域
struct oldest_sequence_num_less_compare {
	bool operator() (const std::pair<std::string, uint64_t> &first,
			 const std::pair<std::string, uint64_t> &second) {
		return first.second < second.second;
	}
};

// 用于删除没用的日志的函数对象，可显著提高代码可读性。以后应多用STL算
// 法。
struct delete_old_log_op {
	void operator() (const std::pair<uint64_t, fs::path> &elem) {
		LOG (TRACE, "deleting old log file "
		     << elem.second
		     << " whose max sequence number is "
		     << elem.first
		     << ".");
		
		int ret = fs::remove(elem.second.string());
		// 删除失败，则记录日志后跳过
		if(ret < 0) {
			LOG(ERROR, "remove old log file "
			    << elem.second
			    << " failed, fs::errno = "
			    << fs::get_errno() << "skip it.");
		}
	}
};

log_thread::log_thread ()
	: m_stop(false)
	, m_log_entry_counter(0)
	, m_log_bytes(0)
	, m_cur_sequence_num(0)
	, m_cur_logfile(fs::BAD_FILE)
	, m_cur_file_seq(0)
	, m_last_roll_back_time(boost::get_system_time())
	, m_rollback_interval(global::the_conf->roll_back_interval)
{
	m_cache_logfile_dir = global::the_conf->cache_logfile_dir;
	if(! local_fs::exists(m_cache_logfile_dir))
	{
		const bool ok = local_fs::mkdir(m_cache_logfile_dir);
		LOG_IF((!ok), ERROR,
		       "create cache logfile dir failed, dir = " << m_cache_logfile_dir
			<< ", stop log thread");
		if(! ok)
		{
			m_stop = true;
			global::stop_request = true;
		}
	}
	m_cache_logfile_dir /= naming_strategy::server_dirname(global::the_address_info->m_hostname,
							     global::the_address_info->m_port,
							     global::the_address_info->m_startcode);
	if(! local_fs::exists(m_cache_logfile_dir))
	{
		const bool ok = local_fs::mkdir(m_cache_logfile_dir);
		LOG_IF((!ok), ERROR,
		       "create cache logfile dir failed, dir = " << m_cache_logfile_dir
		       << ", stop log thread");
		if(! ok) 
		{
			m_stop = true;
			global::stop_request = true;
		}
	}
	m_cache_logfile_seq = cur_micro_seconds();

	m_cur_sequence_num = m_cache_logfile_seq;
	m_cur_file_seq = m_cur_sequence_num.value();

	int ret = this->switch_logfile_no_lock();
	if(ret < 0)		// 初始化失败
	{
		m_stop = true;
		global::stop_request = true;
		LOG(ERROR, "switch_logfile_no_lock failed");
	}
}

// 调用时需要对cur_logfile_lock加锁
int log_thread::switch_logfile_no_lock()
{
	if (m_cur_logfile != fs::BAD_FILE)
	{
		const uint64_t seq_num = m_cur_sequence_num.value() - 1;
		m_history_log.insert(std::multimap<uint64_t, fs::path>::value_type(seq_num, m_cur_logfile_path));
		// 如果有日志文件的seq_num相同了，则记录下来，分析一下是什么原因
		LOG_ENV(const std::size_t debug_count = m_history_log.count(seq_num));
		LOG_IF((debug_count != 1),
		       WARN,
		       "log_thread::switch_logfile_no_lock oldlog file path "
		       << m_cur_logfile_path << " with sequence number " << seq_num
		       << " has " << debug_count << " files");

		fs::close(m_cur_logfile);
		LOG(INFO, "log_thread::switch_logfile_no_lock save and close current logfile "
		    << m_cur_logfile_path << " with sequence number " << seq_num
		    << " to get a new one");
	}

	if(! fs::exists(global::server_root_path()))
	{
		LOG(ERROR, "tablet server dir not exists, quit log thread ");
		m_stop = true;	// 不能持有锁，也无需notify，因为就是自己这个线程
		global::stop_request = true;
		return -1;
	}
	m_cur_logfile_path = global::server_root_path() / boost::lexical_cast<std::string>(m_cur_file_seq);
	m_cur_logfile = fs::create(m_cur_logfile_path.string());
	
	++ m_cur_file_seq;
	// 清除计数器
	m_log_entry_counter = 0;
	m_log_bytes = 0;
	
	if (m_cur_logfile == fs::BAD_FILE)
	{
		LOG(ERROR, "log_thread::switch_logfile create new log file "
		    << m_cur_logfile_path << " failed, fs::errno = "
		    << fs::get_errno() << ", do what?");
		return -1;
	}
	LOG(INFO, "log_thread::switch_logfile new logfile "
	    << m_cur_logfile_path
	    << " created");
	return 0;
}

bool log_thread::log(const std::string &tablet_name,
		    const std::string &row_name,
		    const boost::asio::const_buffer &group_data,
		    const std::vector<std::string> &group_list)
{
	if(m_stop)
	{
		LOG(DEBUG, "log thread stopped");
		return false;
	}
	
	const std::size_t header_size =
		sizeof(uint64_t) * 3 +
		tablet_name.size() + sizeof(uint32_t) +
		row_name.size() + sizeof(uint32_t) +
		sizeof(uint32_t);
	const std::size_t total_size =
		header_size + boost::asio::buffer_size(group_data);

	boost::mutex::scoped_lock lock(m_cache_logfile_mutex);

	if((! m_cache_logfile) ||
	   (m_cache_logfile->avaliable_bytes() < total_size))
	{
		const std::size_t conf_size = global::the_conf->cache_logfile_size;
		const std::size_t bytes = (std::max)(total_size,
						     conf_size);

		++m_cache_logfile_seq;
		m_cache_logfile_path = m_cache_logfile_dir / boost::lexical_cast<std::string>(m_cache_logfile_seq);

		logfile_ptr ptr(new logfile(m_cache_logfile_path.string(),
					    bytes));
		const int ret = ptr->open();
		if(ret < 0)
		{
			ptr.reset();
			local_fs::remove(m_cache_logfile_path);
			return ret;
		}

		if(m_cache_logfile)
		{
			boost::mutex::scoped_lock lock_old(m_old_cache_logfile_mutex);
			const bool need_notify = m_old_cache_logfile.empty();
			m_old_cache_logfile.push_back(m_cache_logfile);
			if(need_notify)
				m_has_old_cache_logfile.notify_one();
			lock_old.unlock();
			m_cache_logfile.reset();
		}
		
		m_cache_logfile = ptr;
	}
	logfile &file = *m_cache_logfile;

	const boost::asio::mutable_buffer buf = file.prepare(total_size);
	if(boost::asio::buffer_size(buf) != total_size)
	{
		return -1;
	}

	kb::binary_oarchive<kb::network_byteorder> out(buf);
	const uint64_t sequence = get_sequence_number();
	const uint64_t timestamp = cur_micro_seconds();
	const uint64_t total_size_to_save = static_cast<uint64_t>(total_size);
	const uint32_t group_data_size = boost::asio::buffer_size(group_data);
	assert(group_data_size <= static_cast<std::size_t>(std::numeric_limits<uint32_t>::max()));
	const uint32_t group_data_size_to_save = static_cast<uint32_t>(group_data_size);
	bool ok = out.save(sequence) &&
		out.save(timestamp) &&
		out.save(total_size_to_save) &&
		out.save(tablet_name) &&
		out.save(row_name) &&
		out.save(group_data_size_to_save);
	if(ok)
	{
		assert(out.left_buffer_size() == boost::asio::buffer_size(group_data));
		::memcpy(out.left_buffer(),
			 boost::asio::buffer_cast<const char*>(group_data),
			 out.left_buffer_size());
		ok = file.commit_log(total_size, sequence);
	}
	lock.unlock();

	// 设置子表名+组名的最小顺序号
	// 直接更新到 m_oldest_sequence_num_for_tablet_group
	if(ok)
	{
		boost::unique_lock<boost::shared_mutex> guard(m_oldest_sequence_num_for_tablet_group_lock);
		BOOST_FOREACH(const std::string &group_name, group_list)
		{
			const std::string tablet_name_with_group_name
				= naming_strategy::composited_tablet_group_name(tablet_name,
										group_name);
			const seq_map::iterator iter
				= m_oldest_sequence_num_for_tablet_group.find(tablet_name_with_group_name);
			// 只保存最小的，所以没有的时候才写入
			if(iter == m_oldest_sequence_num_for_tablet_group.end()) // 只在不存在时记录。因为队列中的日志序列号是有序的，所以这样没问题
			{
				m_oldest_sequence_num_for_tablet_group.insert(seq_map::value_type(tablet_name_with_group_name,
												  sequence));
			}
			else if(iter->second > sequence) // 安全性检查
			{
				LOG(WARN, "oldest sequence is NOT least, oldest = " << iter->second
				    << ", cur = " << sequence
				    << ", so now correct the oldest one!");
				iter->second = sequence;
			}
		}
	}
	
	return ok;
}

// int log_thread::log(log_entry_v2_ptr &entry)
// {
// 	{
// 		boost::mutex::scoped_lock lock(m_queue_lock);

// 		LOG_IF((m_queue.size() > std::size_t(global::the_conf->log_queue_size)),
// 		       WARN,
// 		       "log_thread::log log queue is too large("
// 		       << m_queue.size() << ") against "
// 		       << std::size_t(global::the_conf->log_queue_size));

// 		entry->set_sequence_number(get_sequence_number());
// 		m_queue.push_back(entry);
// 	}
// 	m_queue_not_empty.notify_one();
// 	return 0;
// }

void log_thread::stop()
{
	m_stop = true;
	m_has_old_cache_logfile.notify_all();
}

void log_thread::operator() () {
	LOG(INFO, "log thread started.");

	while (true) {
		boost::mutex::scoped_lock lock(m_old_cache_logfile_mutex);
		// 因为还要检查是否需要回滚日志，所以不能老等在
		// 这里，要用timed_wait，等到下次回滚检查时间
		if(m_old_cache_logfile.empty())
		{
			// 注意：！！！
			// 只在没有日志消息的时候才检查是否退出，使其在
			// 退出前将所有日志消息全部处理完
			if (m_stop)
			{
				LOG(INFO, "log_thread is stopped");
				
				lock.unlock();
				boost::mutex::scoped_lock lock_2(m_cache_logfile_mutex);
				if((m_cache_logfile) &&
				   (m_cache_logfile->log_count() != 0)) // 还有日志在缓冲中
				{
					LOG(INFO, "has cached logfile, flush it now");
					lock.lock();
					m_old_cache_logfile.push_back(m_cache_logfile);
					lock.unlock();
					m_cache_logfile.reset();
				}
				else
				{
					LOG(INFO, "no any log, roll back logfiles and quit");
					roll_back(); // 退出前再加一次强制roll_back()操作，以删除无用日志
					break;
				}

				lock_2.unlock();
				lock.lock();
			}

			// 最多等到下次回滚检查时间
			if(m_old_cache_logfile.empty())
			{
				m_has_old_cache_logfile.timed_wait(lock,
								   m_last_roll_back_time + m_rollback_interval);
			}
		}

		if (! m_old_cache_logfile.empty())
		{
			m_writing_old_cache_logfile.swap(m_old_cache_logfile);
		}
		lock.unlock();

		if(m_writing_old_cache_logfile.empty())
		{
			// 如果 m_writing_old_cache_logfile 是空的，就
			// 说明 m_old_cache_logfile 是空的；因此这一次
			// 运行不是有旧日志通知，而是超时，这时候检查
			// m_cache_logfile 是否有数据，若有则将其加入
			// m_writing_old_cache_logfile 中并写入日志文
			// 件，以避免日志长期保留在 cache logfile 中
			boost::mutex::scoped_lock lock_cache_logfile(m_cache_logfile_mutex);
			if(m_cache_logfile &&
			   (boost::asio::buffer_size(m_cache_logfile->data()) != 0))
			{
				m_writing_old_cache_logfile.push_back(m_cache_logfile);
				m_cache_logfile.reset(); // 直接将其重置，下次记日志时会检测到并创建的
				LOG(DEBUG, "wait log timeout, flush cur cache logfile");
			}
		}

		// 找到所有子表组中最老的日志序列号S_old，若
		// cache_logfile 的最新序列号比S_old更老，则不必写入日
		// 志文件了。
		bool can_delete_all = false;
		std::pair<std::string, uint64_t> oldest_elem;
		// oldest lock region
		{
			boost::shared_lock<boost::shared_mutex>
				guard(m_oldest_sequence_num_for_tablet_group_lock);
			seq_map::iterator
				oldest_iter = std::min_element(m_oldest_sequence_num_for_tablet_group.begin(),
							       m_oldest_sequence_num_for_tablet_group.end(),
							       oldest_sequence_num_less_compare());
			
			// 没有找到，说明该列表为空，则可以删除所有旧日志文件
			if(oldest_iter == m_oldest_sequence_num_for_tablet_group.end())
			{
				can_delete_all = true;
				oldest_elem.second = 0;
			}
			else
			{
				oldest_elem = *oldest_iter;
			}
		}

		int error_times = 0;
		int ret = 0;
		bool skipped = false;
		while(! m_writing_old_cache_logfile.empty())
		{
			logfile_ptr cache_logfile = m_writing_old_cache_logfile.front();
			if(can_delete_all || (cache_logfile->get_largest_sequence_number() < oldest_elem.second))
			{
				LOG(DEBUG, "all logs in cache_logfile has been saved against oldest seqnum"
				    << ", cache_logfile = " << cache_logfile->to_string()
				    << ", can_delete_all = " << std::boolalpha << can_delete_all
				    << ", oldest_seqnum = " << oldest_elem.second
				    << ", oldest_seqnum_tabletgroup = " << oldest_elem.first);
				skipped = true;
				ret = 0; // skip flush
			}
			else
			{
				skipped = false;
				ret = flush_to_logfile(cache_logfile);
			}
			if((! m_stop) && (ret < 0))
			{
				// 如果出错，并发现是子表服务器目录不存在，则会置为stop；否则是未知错误，需要core掉
				++ error_times;
				if(error_times >= global::the_conf->retry_times)
					assert(false);

				// TODO:写日志出错后，首先将日志系统关
				// 闭（使新的记日志操作无法成功），进入
				// 退出状态，再将剩下的日志写到本地文件
				// 系统中的指定文件中（若不是GFS错误，
				// 则写入到GFS中）；完成后向master发送
				// 退出报告，最后才停掉系统。
				//
				// 上面也是所有assert(false)的地方需要
				// 做的操作。
				//
				// 也就是说，需要实现一个系统的退出机制。
			}
			else
			{
				m_writing_old_cache_logfile.pop_front();
				error_times = 0;
				cache_logfile->set_flushed();
				cache_logfile->close();
				LOG(DEBUG, "cache logfile " << cache_logfile->to_string()
				    << " flushed, close and remove it");
				local_fs::remove(cache_logfile->filename());
			}

			if(! skipped)
			{
				// 因为一次写入的数据量可能比较大，所以每写完一个缓冲文件后，加一次roll_back检查

				// 不管有没有日志写入，都检查是否需要回滚。因为还有时间
				// 的因素，很久没写入的话，子表会紧缩，就可以删除所有的
				// 日志了。
				if (need_roll_back())
				{
					const int ret = roll_back();
					LOG_IF((ret < 0), // 回滚错误，只记日志，不作处理。等错误被写日志文件感知到再处理！
					       ERROR, "log_thread:: roll_back error, skip it");
					
					// 虽然回滚可能失败，或根本没进行回滚，
					// 但这里更新时间戳却没有大碍。因为若日
					// 志文件较小，仅是超时，则忽略也没什么
					// 问题；若日志文件较大，那么更新时间也
					// 不会影响回滚
					m_last_roll_back_time = boost::get_system_time(); // update timestamp
				}
			}
		}
			
		// 不管有没有日志写入，都检查是否需要回滚。因为还有时间
		// 的因素，很久没写入的话，子表会紧缩，就可以删除所有的
		// 日志了。
		if (need_roll_back())
		{
			const int ret = roll_back();
			LOG_IF((ret < 0), // 回滚错误，只记日志，不作处理。等错误被写日志文件感知到再处理！
			       ERROR, "log_thread:: roll_back error, skip it");
			
			// 虽然回滚可能失败，或根本没进行回滚，
			// 但这里更新时间戳却没有大碍。因为若日
			// 志文件较小，仅是超时，则忽略也没什么
			// 问题；若日志文件较大，那么更新时间也
			// 不会影响回滚
			m_last_roll_back_time = boost::get_system_time(); // update timestamp
		}
	}

	LOG(INFO, "log thread quit.");
}

int log_thread::flush_to_logfile(logfile_ptr cache_logfile)
{
	boost::mutex::scoped_lock lock(m_cur_logfile_lock);
	const std::size_t max_retry_times =
		global::the_conf->retry_times;
	bool ok = false;
	for(std::size_t retry_count = 0; (! ok) && (retry_count < max_retry_times); ++retry_count)
	{
		if(fs::BAD_FILE == m_cur_logfile)
		{
			LOG(INFO, "log_thread::write_to_file log file is closed, create a new log file");
			int ret = switch_logfile_no_lock();
			if(ret < 0)
			{
				LOG(ERROR, "log_thread::write_to_file create new log file failed, fs::errno = "
				    << fs::get_errno() << ", abort");
				return -1;
			}
		}

		const std::size_t log_count = cache_logfile->log_count();
		const boost::asio::const_buffer data = cache_logfile->data();
		const std::size_t total_bytes = boost::asio::buffer_size(data);
		const fs::ssize_t ret = fs::writen(m_cur_logfile, data);
		if(ret != static_cast<fs::ssize_t>(total_bytes)) // write error
		{
			LOG(ERROR, "log_thread: write "
			    << log_count << " log entry(ies) "
			    << " failed, totally "
			    << total_bytes
			    << " bytes, and return value is "
			    << ret
			    << ", fs::errno = " << fs::get_errno()
			    << ", cur_logfile = " << m_cur_logfile_path
			    << ", cache_logfile = " << cache_logfile->filename()
			    << ", close cur log file to get a new one for retrying");
			fs::close(m_cur_logfile);
			fs::remove(m_cur_logfile_path);
			m_cur_logfile = fs::BAD_FILE; // make new log file to create
			continue; // retry
		}
		// 写成功，则增加计数器
		m_log_entry_counter += log_count;
		m_log_bytes += total_bytes;
		ok = true;
		LOG(DEBUG, "log_thread: " << log_count << " log written");
		break;
	}
	lock.unlock();
	if(! ok)		// 没写成功，返回失败
	{
		return -1;
	}

	// // 为“子表名+组名”记录最小日志序列号
	// seq_map &seq_num_cache = cache_logfile->seq_num_cache();
	// seq_map::iterator s_iter = seq_num_cache.begin();
	// const seq_map::iterator s_iter_end = seq_num_cache.end();

	// boost::unique_lock<boost::shared_mutex> guard(m_oldest_sequence_num_for_tablet_group_lock);
	// for(;s_iter != s_iter_end; ++s_iter)
	// {
	// 	const seq_map::iterator iter
	// 		= m_oldest_sequence_num_for_tablet_group.find(s_iter->first);
	// 	// 只保存最小的，所以没有的时候才写入
	// 	if(iter == m_oldest_sequence_num_for_tablet_group.end())
	// 	{
	// 		m_oldest_sequence_num_for_tablet_group.insert(*s_iter);
	// 		LOG(DEBUG, "log_thread::write_to_file save seq num "
	// 		    << s_iter->second
	// 		    << " for tablet_group "
	// 		    << s_iter->first);
	// 	}
	// }
	// guard.unlock();

	return 0;
}

// 从日志队列中取，直到取完或将达到最大日志条数
//
// TODO: 一次写入一批数据有个大的风险，即一个错误的日志，可能会导致很多
// 数据都写不成功，即将错误放大了。如果这里日志采用了写完后再通知就会好
// 一些，外部写一次失败最多再重试一次，即可返回错误。
//
// 或者改为，缓存按日志块分别组织，直到一个错误日志时，将之前的正确日志
// 全部写入，再返回错误即可。写文件用writev同样可保证减少文件的写入次数
//
// 当前的实现中，写入日志的数据已经经过外部的严格检查，这里尽可能的去掉
// 任何不必需的检查
// int log_thread::write_to_file()
// {
// 	// 清除上次出错可能导致的遗留数据
// 	m_sequence_num_cache.clear();

// 	m_data_vec.clear();
// 	m_log_vec.clear();

// 	std::size_t total_bytes = 0;
// 	std::size_t log_count = 0;
// 	for(;(! m_writing_queue.empty()); ++ log_count)
// 	{
// 		const log_entry_v2_ptr &entry = m_writing_queue.front();
		
// 		// 设置子表名+组名的最小顺序号
// 		const std::string &tablet_name = entry->tablet_name();
// 		const std::vector<std::string> groups = entry->groups();
// 		BOOST_FOREACH(const std::string &group_name, groups)
// 		{
// 			const std::string tablet_name_with_group_name
// 				= naming_strategy::composited_tablet_group_name(tablet_name,
// 										group_name);
// 			const seq_map::iterator iter
// 				= m_sequence_num_cache.find(tablet_name_with_group_name);
// 			if(iter == m_sequence_num_cache.end()) // 只在不存在时记录。因为队列中的日志序列号是有序的，所以这样没问题
// 			{
// 				m_sequence_num_cache.insert(seq_map::value_type(tablet_name_with_group_name,
// 										  entry->sequence_number()));
// 			}
// 		}

// 		// 添加到数据数组中，不拷贝内存
// 		m_log_vec.push_back(entry); // 保存住该Key，延续内存的生命期
// 		m_data_vec.push_back(entry->header_data());
// 		m_data_vec.push_back(entry->body_data());
// 		total_bytes += entry->bytes();

// 		m_writing_queue.pop_front();
// 	}

// 	// 写文件
// 	bool ok = false;	 // 写成功标志
// 	{
// 		boost::mutex::scoped_lock guard(m_cur_logfile_lock);
// 		const int retry_num = 1; // 只重试一次
// 		for(int retry_count = 0; (! ok) && (retry_count <= retry_num); ++retry_count)
// 		{
// 			if(fs::BAD_FILE == m_cur_logfile)
// 			{
// 				LOG(INFO, "log_thread::write_to_file log file is closed, create a new log file");
// 				int ret = switch_logfile_no_lock();
// 				if(ret < 0)
// 				{
// 					LOG(ERROR, "log_thread::write_to_file create new log file failed, fs::errno = "
// 					    << fs::get_errno() << ", abort");
// 					return -1;
// 				}
// 			}
			
// 			// 用writev一次性写入
// 			const fs::ssize_t ret = fs::writen(m_cur_logfile, m_data_vec);
// 			if(ret != static_cast<fs::ssize_t>(total_bytes)) // write error
// 			{
// 				LOG(ERROR, "log_thread::write_to_file write "
// 				    << log_count << " log entry(ies) "
// 				    << " failed, totally "
// 				    << total_bytes
// 				    << " bytes, and return value is "
// 				    << ret
// 				    << ", fs::errno = " << fs::get_errno()
// 				    << ", close cur log file to get a new one for retrying");
// 				fs::close(m_cur_logfile);
// 				m_cur_logfile = fs::BAD_FILE; // make new log file to create
// 				continue; // retry
// 			}
// 			// 写成功，则增加计数器
// 			m_log_entry_counter += log_count;
// 			m_log_bytes += total_bytes;
// 			ok = true;
// 			// 清理，即可释放内存
// 			m_data_vec.clear();
// 			m_log_vec.clear();
// 			LOG(DEBUG, "log_thread::write_to_file " << log_count << " log written");
// 			break;
// 		}
// 	}
// 	if(! ok)		// 没写成功，返回失败
// 	{
// 		return -1;
// 	}

// 	// 为“子表名+组名”记录最小日志序列号
// 	{
// 		boost::unique_lock<boost::shared_mutex> guard(m_oldest_sequence_num_for_tablet_group_lock);

// 		seq_map::iterator s_iter = m_sequence_num_cache.begin();
// 		const seq_map::iterator s_iter_end = m_sequence_num_cache.end();
// 		for(;s_iter != s_iter_end; ++s_iter)
// 		{
// 			const seq_map::iterator iter
// 				= m_oldest_sequence_num_for_tablet_group.find(s_iter->first);
// 			// 只保存最小的，所以没有的时候才写入
// 			if(iter == m_oldest_sequence_num_for_tablet_group.end())
// 			{
// 				m_oldest_sequence_num_for_tablet_group.insert(*s_iter);
// 				LOG(DEBUG, "log_thread::write_to_file save seq num "
// 				    << s_iter->second
// 				    << " for tablet_group "
// 				    << s_iter->first);
// 			}
// 		}
// 	}

// 	// 清理
// 	m_sequence_num_cache.clear();
// 	return 0;
// }

int log_thread::roll_back() {
	// 由于加入了时间因素，所以日志中可能就是没有内容。
	// 必须加在这里，而不能放在need_roll_back()中，这样会导致时间无法更新而总是超时
	if(m_log_bytes > 0)	// 只有当前日志文件中有内容才切换文件，避免切换空文件
	{
		boost::mutex::scoped_lock _guard(m_cur_logfile_lock);
		int ret = switch_logfile_no_lock(); // 切换文件
		if(ret < 0)
		{
			LOG(ERROR, "log_thread::roll_back switch logfile error, abort");
			return -1;
		}
	}

	// 由于次要紧缩的时候会获取该锁，为避免日志线程长时间阻塞，在这
	// 里只进行尝试，没得到则退出这次紧缩。
	//
	// boost::mutex::scoped_lock guard(m_roll_back_lock);
	boost::mutex::scoped_try_lock try_guard(m_roll_back_lock, boost::try_to_lock);
	if(! try_guard.owns_lock())
	{
		LOG(INFO, "log_thread::roll_back get roll back lock failed, skip it");
		return 0;
	}
	
	// 检查是否有旧的日志文件可以被删除
	
	// history_logs不需要加锁，只这一个线程会用到
	if (m_history_log.size () > 0) { // 有旧日志文件
		LOG(TRACE, "ready to check deletion for history logfiles("
		    << m_history_log.size() << ").");
		
		// 遍历找出所有子表中最老的序列号
		bool can_delete_all = false;
		std::pair<std::string, uint64_t> oldest_elem;
		// oldest lock region
		{
			//Guard<ReadWriteLock::ReadLock>
			boost::shared_lock<boost::shared_mutex>
				guard(m_oldest_sequence_num_for_tablet_group_lock);
			seq_map::iterator
				oldest_iter = std::min_element(m_oldest_sequence_num_for_tablet_group.begin(),
							       m_oldest_sequence_num_for_tablet_group.end(),
							       oldest_sequence_num_less_compare());
			
			// 没有找到，说明该列表为空，则可以删除所有旧日志文件
			if(oldest_iter == m_oldest_sequence_num_for_tablet_group.end())
			{
				can_delete_all = true;
				oldest_elem.second = 0;
			}
			else
			{
				oldest_elem = *oldest_iter;
			}
		}
		
		// 在这里所有的旧日志都可以被删除
		delete_old_log_op delete_op; // 生成删除函数
		if (can_delete_all) { // 没有新写入的日志
			LOG(DEBUG, "no new log entry written, deleting all old log file.");
			
			std::for_each(m_history_log.begin(),
				      m_history_log.end(),
				      delete_op);
			// 清空列表
			m_history_log.clear();
		} else {
			LOG(TRACE, "found oldest sequence number "
			    << oldest_elem.second
			    << " which belongs to tablet_group "
			    << oldest_elem.first);
			
			// 找出所有ID小于oldest_elem的日志文件
			std::multimap<uint64_t, fs::path>::iterator
				end_pos = m_history_log.lower_bound (oldest_elem.second);
			std::multimap<uint64_t, fs::path>::iterator
				begin_pos = m_history_log.begin();
			
			if (begin_pos != end_pos)
			{
				// 这时，从begin()到end_pos指向的位置之
				// 前，即[begin(), end_pos)，这里的所有
				// 日志文件全部是可以被删除的
				std::for_each(begin_pos, end_pos, delete_op);
				
				// 将删除的文件从历史文件列表中清除
				m_history_log.erase (begin_pos, end_pos);
				
				LOG(INFO, "some old logfiles have been deleted");
			}
			else
			{
				LOG(DEBUG, "no old logfiles need to delete");
			}
		}
	} else {
		LOG(TRACE, "no old logfiles in history list");
	}
	return 0;
}

// 按字节数来确定
bool log_thread::need_roll_back () const {
	return (m_log_bytes >= global::the_conf->max_log_bytes)
		|| ((boost::get_system_time() - m_last_roll_back_time) >= m_rollback_interval);
}

// 次要紧缩完成后，调用该函数删除子表名列表中的条目。此函数必须在持有回
// 滚锁的时候才能调用，所以无需加锁。
void log_thread::complete_minor_compact (const std::string &tablet_group,
					 uint64_t seq_num) {
	//Guard<ReadWriteLock::WriteLock>
	boost::unique_lock<boost::shared_mutex> guard(m_oldest_sequence_num_for_tablet_group_lock);

	const seq_map::iterator
		it = m_oldest_sequence_num_for_tablet_group.find(tablet_group);

	// 若紧缩的seq_num大于等于已有的最老ID，则说明它已经被写到文件
	// 了，这时将其删除。在次要紧缩期间，虽会有新日志记录，但只会写
	// 到当前日志文件中，并且文件不会改变，这时记下的最老ID虽不是真
	// 正的最老ID，但却和最老ID同属于一个日志文件，一样能保证当前日
	// 志文件不被删除。
	//
	// 这里不能改为更新seq_num为最新的，否则一个不再写数据的子表会
	// 使整个日志无法删除。
	if (it != m_oldest_sequence_num_for_tablet_group.end()) { // 已有
		LOG(DEBUG, "found oldest seq num "
		    << it->second
		    << " for tablet_group "
		    << tablet_group
		    << " with minor compact seq num "
		    << seq_num
		    << ".");
		if (seq_num >= it->second) { // 且大于等于
			m_oldest_sequence_num_for_tablet_group.erase(it);
			LOG(DEBUG, "the oldest seq_num "
			    << it->second << " deleted against to " << seq_num
			    << " in " << tablet_group);
		} else {
			LOG(DEBUG, "the oldest seq num " << it->second << " for "
			    << tablet_group << " unchanged");
		}
	}

	/*
	  seq_map::iterator
	  it = _oldest_sequence_num_for_tablet.find(tablet_name);

	  // 仅当没有更新的写入时才能删除，若有更新的则必须保留
	  if ((it != _oldest_sequence_num_for_tablet.end())
	  && (seq_num >= it->second))
	  _oldest_sequence_num_for_tablet.erase(it);
	*/
}
